/*
 * Hyuga: A Density-Grid Stream Clustering Platform. Copyright (C) 2014 PUC-Rio/Laboratory for Advanced Collaboration
 * 
 * Hyuga is free software: you can redistribute it and/or modify it under the terms of the GNU General Public License as
 * published by the Free Software Foundation, either version 3 of the License, or (at your option) any later version.
 * 
 * Hyuga is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty
 * of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details.
 * 
 * You should have received a copy of the GNU General Public License along with Hyuga. If not, see
 * <http://www.gnu.org/licenses/>.
 */
package br.pucrio.inf.lac.konoha.hyuga.playground;


import java.io.IOException;
import java.util.Collection;

import org.apache.log4j.ConsoleAppender;
import org.apache.log4j.DailyRollingFileAppender;
import org.apache.log4j.Level;
import org.apache.log4j.Logger;
import org.apache.log4j.PatternLayout;

import com.espertech.esper.client.Configuration;
import com.espertech.esper.client.ConfigurationEngineDefaults.Threading;
import com.espertech.esper.client.EPServiceProvider;
import com.espertech.esper.client.EPServiceProviderManager;

/**
 * A CSV playback routine. The csv file must have the following columns: uuid,latitude,longitude,timestamp
 * 
 * Sample File: uuid, latitude, longitude, timestamp 286, -22.9559986550780, -43.1922377359216, 70462 973,
 * -22.9603834270102, -43.2000845051359, 70462 2742, -22.9728237658448, -43.1907249148011, 704637 4075,
 * -22.9788327380116, -43.1908675079309, 70464 878, -22.9811741072085, -43.2333375518404, 70464 3431, -22.9703160507337,
 * -43.1907072643957, 704650 341, -22.9656428061742, -43.2163599814267, 70466 5028, -22.9580526401979,
 * -43.2050608584764, 704664
 *
 * @author Marcos Roriz <mroriz@inf.puc-rio.br>
 */
public class PlaygroundMain {

    /** LOGGER. */
    private static Logger     LOGGER = Logger.getLogger("br.pucrio.inf.lac.konoha.hyuga");

    /** CEP ENgine. */
    private EPServiceProvider cepEngine;

    private long start;

    private static final Object lock = new Object();

    /**
     * Main Routine.
     * 
     * @param args a CSV filename.
     * @throws IOException If log files cannot be created.
     * @throws IllegalAccessException if args is null.
     */
    public static void main(String[] args) throws Exception {
        // Logging
        DailyRollingFileAppender fileAppender = new DailyRollingFileAppender(new PatternLayout("[%-4p] %d %c - %m%n"),
                                                                             "play-log", "'.'yyyy-MM-dd");
        ConsoleAppender consoleAppender = new ConsoleAppender(new PatternLayout("[%-4p] [%c] %m%n"));

        Logger.getRootLogger().setLevel(Level.INFO);
        Logger.getLogger("br.pucrio.inf.lac").addAppender(fileAppender);
        Logger.getLogger("br.pucrio.inf.lac").addAppender(consoleAppender);
        Logger.getLogger("br.pucrio.inf.lac").setLevel(Level.INFO);
        Logger.getLogger("com.espertech.esper").addAppender(consoleAppender);

        LOGGER.info("CSV Playback...");

        // CSV
        PlaygroundMain play = new PlaygroundMain();
        play.setupCEP(play);
        play.play();
    }

    private void play() throws Exception {
        start = System.currentTimeMillis();
        cepEngine.getEPRuntime().sendEvent(new A(3));
        cepEngine.getEPRuntime().sendEvent(new A(3));
        cepEngine.getEPRuntime().sendEvent(new A(10));
        cepEngine.getEPRuntime().sendEvent(new A(3));
        
        int v[] = new int[100];
        for (int i = 0; i < 100; i++) {
            v[i] = i;
        }
        
        for (int i = 0; i < 100; i++) {
            final int x = v[i]; 
            new Thread (new Runnable() {
                @Override
                public void run() {
                    cepEngine.getEPRuntime().sendEvent(new A(x));
                }
            }).start();;
        }
        
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
        String query = 
                "select * from K";
        Object k = cepEngine.getEPRuntime().executeQuery(query);
        System.out.println("terminei de mandar evt");
    }

    public void setupCEP(Object subscriber) {
        if (subscriber == null) throw new NullPointerException();

        // CEP Config
        Configuration cepConfig = new Configuration();

        // Threading and Tuning
        Threading threading = cepConfig.getEngineDefaults().getThreading();
//        threading.setInsertIntoDispatchTimeout(5000);
//        threading.setListenerDispatchTimeout(5000);
        threading.setThreadPoolInbound(true);
        threading.setThreadPoolInboundNumThreads(16);
//        threading.setThreadPoolOutbound(false);
//        threading.setThreadPoolOutboundNumThreads(16);
        cepConfig.getEngineDefaults().getViewResources().setShareViews(false);

        // Event Types
        cepConfig.addEventType("A", A.class.getName());
        cepConfig.addEventType("B", B.class.getName());
        cepConfig.addEventType("Buffer", Buffer.class.getName());
        cepConfig.addEventType("C", C.class.getName());
        cepConfig.addEventType("D", D.class.getName());
        
        cepConfig.addPlugInSingleRowFunction("aff",   Helper.class.getCanonicalName(), "aff");
        cepConfig.addPlugInSingleRowFunction("m",   Helper.class.getCanonicalName(), "m");
        cepConfig.addPlugInSingleRowFunction("zlock", Helper.class.getCanonicalName(), "zlock");
        cepConfig.addPlugInSingleRowFunction("zunlock", Helper.class.getCanonicalName(), "zunlock");
        

        // CEP
        cepEngine = EPServiceProviderManager.getProvider("ESPER", cepConfig);
        
        // EPL        
        String window    = " CREATE WINDOW  K.win:time(" + (120) + " sec) AS B";
        String window2   = " CREATE WINDOW  F.win:time(" + (120) + " sec) AS B";
        String ctx = "CREATE CONTEXT Cell PARTITION BY x FROM A";
        
        String upserEPA = " @Name('FFF') ON A"
                        + " MERGE INTO K"
                        + " WHERE A.x < K.y"
                        + " WHEN MATCHED "
                        + "      THEN UPDATE SET K.y = A.x"
                        + " WHEN NOT MATCHED "
//                        + "      THEN UPDATE SET B.lock = plock()"
//                        + "      THEN INSERT INTO K SELECT A.x AS y"
                        + "      THEN INSERT INTO Buffer SELECT K.y AS y";
//                        + "      THEN UPDATE SET plock()"         
        
        String bufferEPA = " INSERT INTO K"
                         + " SELECT x as y"
                         + " FROM A";
//                         + " WHERE zlock()";
        
        
        String selDelEPA = " @Name('SELDEL') ON C "
                         + " INSERT INTO K "
                         + " SELECT AND DELETE aff(window(winName.*).selectFrom(v => v)) AS y"
                         + " FROM K as winName"
                         + " WHERE K.y > -50";
        
        String debugEPA = " @Name('Debug') @Audit"
//                        + " CONTEXT Cell"
                        + " ON K INSERT INTO B"
                        + " SELECT AND DELETE m(window(W.*))"
                        + " FROM K as W";
//
//                + " INSERT ISTREAM INTO D"
//                + " SELECT "
//                + "        ( SELECT * "
//                + "          FROM A.win:time(3 sec) AS ev"
//                + "          WHERE ev.x = LU.x).selectFrom(v => v) as z"
//                + " FROM A as LU";

        String dEPA = "@Name('Outro') SELECT ISTREAM * FROM B";
        cepEngine.getEPAdministrator().createEPL(ctx);

        cepEngine.getEPAdministrator().createEPL(window);
        cepEngine.getEPAdministrator().createEPL(window2);
        cepEngine.getEPAdministrator().createEPL(bufferEPA);
//        cepEngine.getEPAdministrator().createEPL(upserEPA);
//        cepEngine.getEPAdministrator().createEPL(selDelEPA).setSubscriber(this);
        cepEngine.getEPAdministrator().createEPL(debugEPA).setSubscriber(this);
        cepEngine.getEPAdministrator().createEPL(dEPA).setSubscriber(this);
    }
    
    public void update(Collection<?> a) {
        System.out.println("---- OUTPUT DO INSERT + " + a.size());
        System.out.println("time: " + (System.currentTimeMillis() - start) + " msseconds");
        
//        Helper.zunlock();
    }
    
    public void update(A[] l) {
        System.out.println("---- OUTPUT DO INSERT + ");
        System.out.println("time: " + (System.currentTimeMillis() - start) + " msseconds");
        
//        Helper.zunlock();
    }
    
    public void update(D l) {
        System.out.println("---- CHEGOU NO D " + l);
        System.out.println("time: " + (System.currentTimeMillis() - start) + " msseconds");
//        Helper.zunlock();
    }
    
}
